from __future__ import print_function, division
from pyspark import SparkConf, SparkContext
from pyspark.sql import SparkSession

# 转换栏位的属性
from pyspark.sql.types import DoubleType
# 表示在本地读取  任务名称test
spark = SparkSession.builder.master("local") \
    .appName("test") \
    .enableHiveSupport() \
    .getOrCreate()

# 使用sparkContext操作上下文
sc = spark.sparkContext

######################################

# 读取文件  读取csv文件 ，文件以英文逗号做分割
fileDF = spark.read.csv("pysprk_data/ratings.csv", sep=",", header=True)

# 查看文件
# fileDF.show()
print("查看文档中的所有列栏位(colums)：{}".format(fileDF.columns))

print("查看文档中列的长度：{}".format(len(fileDF.columns)))

print("查看栏位统计值(descride())")
fileDF.describe().show()
print("查看userid列的统计数据")
fileDF.describe('userid').show()
print("查看该对象所有栏位对应的")
fileDF.printSchema()

print("选择栏位select()")
fileDF.select("userid",'rating').show()

print("查看两列，其中一列的值加1")
fileDF.select('userid', fileDF['rating'] + 1).show()

print("再次查看选择列的类型，其中计算过的列变为double")
fileDF.select('userid', fileDF['rating'] + 1).printSchema()

print("列出所偶userid==3的数据")
fileDF.filter(fileDF['userid'] == 3).show()

print("选择userid==3的数据，并且查看userid，和rating列的数据")
fileDF.filter(fileDF['userid'] == 3).select('userid', 'rating').show()

print("计算不重复值（distinct()表示整行都一样）")
fileDF.select('userid').distinct().show()

print("查看userid列不重复的值")
fileDF.select('userid').distinct().show()

#######################################
# 查看有几部电影评分超过5分
fileDF.filter(fileDF['rating']==5).select('movieid').distinct().count()

#######################################
# 观察数值的分配
print("查看数值的分配")
fileDF.crosstab('userid', 'rating').show()

#######################################
print("给没有数据空补充值为0")
fileDF.fillna('0').show()

#########################################

print("增加一类为rating_double,数据位rating列中的数据，然后改变类型为double")
fileDF = fileDF.withColumn("rating_double", fileDF["rating"].cast(DoubleType()))

# 去掉没有数据的值
fileDF.dropna().show()

#############################################

# 处理重复值
print("查看userid中每个movieid评论的次数")
fileDF.crosstab("userid", "movieid").show()

print("总共的评论次数减去，去重之后的电影数")
print(fileDF.select("userid", "movieid", "rating").count() -
    fileDF.select("userid", "movieid", "rating").distinct().count())


#########################################
# 直接去重复发现灭有去重
fileDF.dropDuplicates().orderBy(['userid', 'movieid', 'rating'], ascending=[1,0]).show()

# 通过dropDuplicates指定数据相同的列，然后按这个条件去除重复的数据
fileDF.dropDuplicates(['userid', 'movieid', 'rating']).orderBy(['userid', 'movieid', 'rating'], ascending=[1,0]).show()

######################################
# 查看去重复数据
fileDF_nodup = fileDF.dropDuplicates(['userid', 'movieid', 'rating']).orderBy(['userid', 'movieid', 'rating'], ascending=[1,0])
print("查看同一个用户对同一部电影有不同的评论数据：{}".format(fileDF_nodup.select("userid", "movieid", "rating").count() -
    fileDF_nodup.select("userid", "movieid", "rating").distinct().count()))

###############
spark.stop()